package com.ftg.learn.fn.service;

import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.context.AnalysisContext;
import com.alibaba.excel.read.listener.ReadListener;
import com.alibaba.excel.util.ListUtils;
import com.alibaba.excel.util.StringUtils;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ftg.learn.fn.comm.StreamConst;
import com.ftg.learn.fn.entity.Student;
import com.ftg.learn.util.R;
import com.google.gson.Gson;
import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.google.gson.JsonParser;
import com.google.gson.reflect.TypeToken;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.QueryTimeoutException;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author 2234
 */
@Service
@Slf4j
public class ReadService {

    ExecutorService task = Executors.newFixedThreadPool(5);

    @Resource
    RedisTemplate<String, String> redisLockTemplate;

    @Resource
    JdbcTemplate jdbcTemplate;

    @Resource
    ObjectMapper objectMapper;

    /**
     * 初始化方法，主要用于设置线程池、检查并初始化Stream、注册消费者以及设置速率限制和监控指标
     * 该方法在对象创建时自动调用，确保所有必要的资源和配置都被正确设置
     */
    @PostConstruct
    public void init() {
        try {
            // 1. 检查Stream是否存在
            if (!isStreamExists()) {
                log.info("Stream {} does not exist, creating...", StreamConst.STREAM_KEY);
                createStreamWithGroup();
            } else {
                log.info("Stream {} already exists", StreamConst.STREAM_KEY);
                createGroupIfNotExists();
            }
            // 2. 注册消费者
            registerConsumer();
        } catch (Exception e) {
            e.printStackTrace();
            log.error("Stream initialization failed", e);
            throw new RuntimeException("Stream initialization failed", e);
        }
    }

    /**
     * 判断Redis中的Stream是否存在
     * 该方法尝试通过调用redisLockTemplate的opsForStream().info()方法来检查指定的Stream键是否存在
     * 如果Stream键存在，则返回true；如果不存在，则返回false
     * 此方法主要用于在执行Stream相关操作前，确认Stream是否已经创建
     *
     * @return boolean 表示Stream是否存在的布尔值，存在返回true，否则返回false
     */
    private boolean isStreamExists() {
        try {
            redisLockTemplate.opsForStream().info(StreamConst.STREAM_KEY);
            return true;
        } catch (Exception e) {
            if (e.getMessage().contains("no such key") || e.getMessage().contains("ERR no such key")) {
                return false;
            }
            return false;
        }
    }

    /**
     * 创建带有消费者组的Stream
     * 此方法首先在Stream中添加一个初始记录，然后创建一个消费者组，并最后删除这个初始记录
     */
    private void createStreamWithGroup() {
        RecordId initialId = redisLockTemplate.opsForStream()
                .add(StreamRecords.newRecord()
                        .ofObject("initial_data")
                        .withStreamKey(StreamConst.STREAM_KEY));

        // 创建消费者组（从起始位置消费）
        redisLockTemplate.opsForStream().createGroup(StreamConst.STREAM_KEY, ReadOffset.from("0-0"), StreamConst.GROUP_NAME);
        log.info("Created stream {} with initial group {}", StreamConst.STREAM_KEY, StreamConst.GROUP_NAME);

        // 删除初始测试数据
        redisLockTemplate.opsForStream().delete(StreamConst.STREAM_KEY, initialId);
    }

    /**
     * 如果消费者组不存在，则创建消费者组
     * 此方法首先检查指定的流中是否存在具有指定名称的消费者组
     * 如果消费者组不存在，它将创建一个新的消费者组，并记录创建信息
     * 如果消费者组已经存在，则记录该信息
     * 在检查或创建消费者组过程中遇到的任何异常都将被捕获并记录，然后重新抛出
     */
    private void createGroupIfNotExists() {
        try {
            // 获取流中的消费者组信息
            StreamInfo.XInfoGroups groups = redisLockTemplate.opsForStream().groups(StreamConst.STREAM_KEY);

            // 检查是否存在具有指定名称的消费者组
            boolean groupExists = groups.stream()
                    .anyMatch(g -> StreamConst.GROUP_NAME.equals(g.groupName()));

            // 如果消费者组不存在，则创建它
            if (!groupExists) {
                redisLockTemplate.opsForStream().createGroup(StreamConst.STREAM_KEY, StreamConst.GROUP_NAME);
                log.info("Created consumer group {} on existing stream {}", StreamConst.GROUP_NAME, StreamConst.STREAM_KEY);
            } else {
                log.info("Consumer group {} already exists", StreamConst.GROUP_NAME);
            }
        } catch (Exception e) {
            // 记录在检查或创建消费者组过程中遇到的异常
            log.error("Failed to create group", e);
            // 重新抛出异常，以便可以在调用此方法的代码中处理
            throw e;
        }
    }

    /**
     * 注册消费者到Redis流
     * <p>
     * 此方法检查指定的消费者是否已存在于Redis流的消费者组中如果不存在，则通过执行PING命令隐式注册该消费者
     * 隐式注册是因为Redis没有直接的命令来注册消费者，但执行某些命令（如PING）会自动在服务器上注册该消费者
     */
    private void registerConsumer() {
        try {
            // 获取指定流和消费者组的消费者信息
            StreamInfo.XInfoConsumers consumers = redisLockTemplate.opsForStream()
                    .consumers(StreamConst.STREAM_KEY, StreamConst.GROUP_NAME);

            // 检查消费者是否已经存在
            boolean consumerExists = consumers.stream()
                    .anyMatch(c -> StreamConst.CONSUMER_NAME.equals(c.consumerName()));

            // 如果消费者不存在，则执行PING命令以隐式注册消费者
            if (!consumerExists) {
                redisLockTemplate.opsForStream()
                        .pending(StreamConst.STREAM_KEY, StreamConst.GROUP_NAME);
                log.info("Registered new consumer: {}", StreamConst.CONSUMER_NAME);
            }
        } catch (Exception e) {
            // 捕获并记录消费者注册检查过程中的异常
            log.warn("Consumer registration check failed", e);
        }
    }


    @Retryable(value = {QueryTimeoutException.class},
            maxAttempts = 3,
            backoff = @Backoff(delay = 100, maxDelay = 1000))
    private void writeToStream(String json) {
        MapRecord<String, String, String> record = StreamRecords.newRecord()
                .ofMap(Collections.singletonMap("data", json))
                .withStreamKey(StreamConst.STREAM_KEY)
                .withId(RecordId.autoGenerate());
        // 使用Redis模板将记录添加到流中
        redisLockTemplate.opsForStream()
                .add(record);
    }


    /**
     * 读取Excel文件
     *
     * @param path 地址
     * @return
     */
    public R<String> readExcel(String path) {

        EasyExcel.read(path, Student.class, new ReadListener<Student>() {

            public static final int BATCH_COUNT = 10;

            private final List<Student> cachedDataList = ListUtils.newArrayListWithExpectedSize(BATCH_COUNT);
            private final Gson gson = new Gson();

            @Override
            public void invoke(Student o, AnalysisContext analysisContext) {

                if (!StringUtils.isEmpty(o.getId()) && !StringUtils.isEmpty(o.getSno())) {
                    if (cachedDataList.size() >= BATCH_COUNT) {
                        List<Student> newList = new ArrayList<>(cachedDataList);
                        task.execute(() -> {
                            String jsonString = gson.toJson(newList);
                            writeToStream(jsonString);
                        });
                        cachedDataList.clear();
                    }
                    cachedDataList.add(o);
                }
            }

            @Override
            public void doAfterAllAnalysed(AnalysisContext analysisContext) {
                task.execute(() -> {
                    String jsonString = gson.toJson(cachedDataList);
                    writeToStream(jsonString);
                });
                cachedDataList.clear();
            }
        }).sheet(0).headRowNumber(5).doRead();

        return R.ok("success");
    }


    /**
     * 读取redis流 定时读取
     */
    @Scheduled(fixedDelay = 10000)
    public void readStream() {
        //读取redis流 - 消费者
        List<MapRecord<String, Object, Object>> read = redisLockTemplate.opsForStream()
                .read(Consumer.from(StreamConst.GROUP_NAME, StreamConst.CONSUMER_NAME),
                        StreamOffset.create(StreamConst.STREAM_KEY, ReadOffset.lastConsumed()));

        System.out.println("数据列表==>> " + read);
        if (Objects.nonNull(read)) {
            Gson gson = new Gson();
            List<Student> data = read.stream().flatMap(r -> {
                        // 判断 data 是否为 JSON 数组
                        String jsonData = r.getValue().get("data").toString();
                        JsonElement jsonElement = JsonParser.parseString(jsonData);

                        if (jsonElement.isJsonArray()) {
                            JsonArray jsonArray = jsonElement.getAsJsonArray();
                            List<Student> studentList = gson.fromJson(jsonElement, new TypeToken<List<Student>>() {
                            }.getType());
                            return studentList.stream();
                        } else {
                            return Stream.empty();
                        }

                    }).filter(Objects::nonNull)
                    .toList();

            try{
                processData(() ->bathInsertIntoSql(data));
            }catch (Exception e){
                //写死信对队
                writeToDeahLetter(data);
            }
        }
    }

    private void writeToDeahLetter(List<Student> failedBatch) {
        // 记录批次写入失败的日志信息

        // 将失败记录写入死信队列
        List<ObjectRecord<String, String>> deadLetters = failedBatch.stream()
                .map(record -> {
                    try {
                        // 序列化失败的记录，并创建死信队列记录

                        return StreamRecords.newRecord()
                                .ofObject(objectMapper.writeValueAsString(record))
                                .withStreamKey("dead_letter_stream");
                    } catch (JsonProcessingException ex) {
                        // 如果序列化失败，记录错误日志
                        log.error("死信序列化失败", ex);
                        return null;
                    }
                })
                .filter(Objects::nonNull) // 过滤掉序列化失败的记录
                .collect(Collectors.toList());

        // 如果存在有效的死信记录，则将其添加到Redis的死信队列中，并更新监控指标
        if (!deadLetters.isEmpty()) {
            deadLetters.forEach(record -> {
                redisLockTemplate.opsForStream().add(record);
            });
        }
    }

    private Object bathInsertIntoSql(List<Student> data) {
        return jdbcTemplate.batchUpdate("insert into excel_student(id,sno,name,sex,clazz) values (?,?,?,?,?)",new BatchPreparedStatementSetter(){

            @Override
            public void setValues(PreparedStatement ps, int i) throws SQLException {
                ps.setObject(1,data.get(i).getId());
                ps.setObject(2,data.get(i).getSno());
                ps.setObject(3,data.get(i).getName());
                ps.setObject(4,data.get(i).getSex());
                ps.setObject(5,data.get(i).getClazz());
                ps.addBatch();
            }

            @Override
            public int getBatchSize() {
                return data.size();
            }
        });
    }

    private <T> T processData(Callable<T> task) {
        int retryCount = 0;
        while (retryCount <= StreamConst.RETRY_NUM) {
            try {
                return task.call();
            } catch (Exception e) {
                try {
                    Thread.sleep(cacledBackOff());
                    retryCount++;
                } catch (InterruptedException ex) {
                    throw new RuntimeException(ex);
                }
            }
        }
        throw new RuntimeException("Failed to process data after " + StreamConst.RETRY_NUM + " retries.");
    }

    private long cacledBackOff() {
        return (long) Math.pow(2, StreamConst.RETRY_NUM) * 2000;
    }


}
